package com.hyl.msg;

import java.io.UnsupportedEncodingException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
/**
 * 
 * <p>@className:  TransactionListenerImpl</p> 
 * <p>Description: 默认rocketmq 发布端的 事务发布时的 处理接口 </p>  
 * @author: zoudyaou qq:37798955
 * @date:   2018年10月11日 下午2:12:19     
 * @copyright: 2018 zoudayou Inc. All rights reserved. 
 * 注意：代码版权归邹达有所有，任何人未经授权不得私自用于商业目的
 */
public class RockTransactionListenerImpl implements TransactionListener {
	private AtomicInteger transactionIndex = new AtomicInteger(0);
	private ConcurrentHashMap<String, Integer> localTrans = new ConcurrentHashMap<>();
	
	/**因为transaction的流程下，
	 * rocketmq会先发送一个consumer不可见的消息，然后在调用
	TransactionListener这个接口中的executeLocalTransaction方法执行事务，
	然后方法内部需要返回一个LocalTransactionState的枚举信息
	COMMIT_MESSAGE, // 提交
    ROLLBACK_MESSAGE, // 回滚
    UNKNOW, // 未知
         相应的当我们返回的是COMMIT_MESSAGE时，那么producer会把消息提交到mq上
	 */
	@Override
	public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
		//从0开始 自增id
		   int value = transactionIndex.getAndIncrement();
           int status = value % 3;
           localTrans.put(msg.getTransactionId(), status);
           return LocalTransactionState.UNKNOW;
	}

	//需要注意的是checkLocalTransaction是用作mq长时间没有收到
	//producer的executeLocalTransaction响应的时候调用的，
	//这个类在3.0之后的版本就被阉割了，只有接口，却没有实现，那么直接写一个空实现即可，
	@Override
	public LocalTransactionState checkLocalTransaction(MessageExt msg) {
		  Integer status = localTrans.get(msg.getTransactionId());
          if (null != status) {
              switch (status) {
                  case 0:
                      return LocalTransactionState.UNKNOW;
                  case 1:
                      return LocalTransactionState.COMMIT_MESSAGE;
                  case 2:
                      return LocalTransactionState.ROLLBACK_MESSAGE;
              }
          }
          return LocalTransactionState.COMMIT_MESSAGE;
	}
}
